/*
 * $Id$
 * --------------------------------------------------------------------------------------
 * Copyright (c) MuleSoft, Inc.  All rights reserved.  http://www.mulesoft.com
 *
 * The software in this package is published under the terms of the CPAL v1.0
 * license, a copy of which has been included with this distribution in the
 * LICENSE.txt file.
 */

package org.mule.transport.jpa;

import java.util.ArrayList;
import java.util.List;

import javax.persistence.EntityManager;
import javax.persistence.EntityTransaction;
import javax.persistence.Query;

import org.mule.transaction.TransactionCoordination;
import org.mule.transport.ConnectException;
import org.mule.transport.TransactedPollingMessageReceiver;
import org.mule.api.MuleMessage;
import org.mule.api.construct.FlowConstruct;
import org.mule.api.endpoint.InboundEndpoint;
import org.mule.api.lifecycle.CreateException;
import org.mule.api.transaction.Transaction;
import org.mule.api.transport.Connector;

/**
 * <code>JpaMessageReceiver</code> TODO document
 */
public class JpaMessageReceiver extends TransactedPollingMessageReceiver {

	protected JpaConnector jpaConnector;

	private List<?> paramNames;

	private String queryString;

	public static final String POLLING_FREQUENCY = "pollingFrequency";

	/*
	 * For general guidelines on writing transports see
	 * http://www.mulesoft.org/documentation
	 * /display/MULE3USER/Creating+Transports
	 */

	public JpaMessageReceiver(Connector connector, FlowConstruct flowConstruct, InboundEndpoint endpoint) throws CreateException {
		super(connector, flowConstruct, endpoint);
		this.jpaConnector = (JpaConnector) connector;
		parseStatement(jpaConnector.getStatement(endpoint));
		if (endpoint.getProperty(POLLING_FREQUENCY) != null) {
			setFrequency(Long.parseLong(endpoint.getProperty(POLLING_FREQUENCY).toString()));
		}

	}

	protected void parseStatement(String statement) {
		this.paramNames = new ArrayList<Object>();
		this.queryString = this.jpaConnector.parseStatement(statement, this.paramNames);
	}

	@Override
	public void doConnect() throws ConnectException {
		/*
		 * IMPLEMENTATION NOTE: This method should make a connection to the
		 * underlying transport i.e. connect to a socket or register a soap
		 * service. When there is no connection to be made this method should be
		 * used to check that resources are available. For example the
		 * FileMessageReceiver checks that the directories it will be using are
		 * available and readable. The MessageReceiver should remain in a
		 * 'stopped' state even after the doConnect() method is called. This
		 * means that a connection has been made but no events will be received
		 * until the start() method is called.
		 * 
		 * Calling start() on the MessageReceiver will call doConnect() if the
		 * receiver hasn't connected already.
		 */

		/*
		 * IMPLEMENTATION NOTE: If you need to spawn any threads such as worker
		 * threads for this receiver you can schedule a worker thread with the
		 * work manager i.e.
		 * 
		 * getWorkManager().scheduleWork(worker, WorkManager.INDEFINITE, null,
		 * null); Where 'worker' implemments javax.resource.spi.work.Work
		 */

		/*
		 * IMPLEMENTATION NOTE: When throwing an exception from this method you
		 * need to throw an ConnectException that accepts a Message, a cause
		 * exception and a reference to this MessageReceiver i.e.
		 * 
		 * throw new ConnectException(new
		 * Message(Messages.FAILED_TO_SCHEDULE_WORK), e, this);
		 */

	}

	@Override
	public void doDisconnect() throws ConnectException {
		/*
		 * IMPLEMENTATION NOTE: Disconnects and tidies up any rources allocted
		 * using the doConnect() method. This method should return the
		 * MessageReceiver into a disconnected state so that it can be connected
		 * again using the doConnect() method.
		 */

		// TODO release any resources here
	}

	@Override
	public void doDispose() {
		// Optional; does not need to be implemented. Delete if not required

		/*
		 * IMPLEMENTATION NOTE: Is called when the Conector is being dispoed and
		 * should clean up any resources. The doStop() and doDisconnect()
		 * methods will be called implicitly when this method is called.
		 */
	}

	@Override
	protected List<MuleMessage> getMessages() throws Exception {
		if (!flowConstruct.getMuleContext().isPrimaryPollingInstance()) {
			return null;
		}
		EntityManager em = null;
		EntityTransaction etx = null;
		try {
			Transaction tx = TransactionCoordination.getInstance().getTransaction();
			em = jpaConnector.getEntityManager();
			if (tx == null) {
				etx = em.getTransaction();
			}
			if (!em.isOpen()) {
				throw new Exception("Cannot open entityManager.");
			}
			if (etx != null && !etx.isActive()) {
				etx.begin();
			}
			List<?> resultSet = null;
			Query query = em.createQuery(queryString);
			resultSet = query.getResultList();
			List<MuleMessage> result = new ArrayList<MuleMessage>();
			for (Object e : resultSet) {
				result.add(createMuleMessage(e, endpoint.getEncoding()));
			}
			if (etx != null && etx.isActive()) {
				etx.commit();
			}
			return result;
		} catch (Exception e) {
			logger.error("Error querying with JPA Transport", e);
			if (etx != null && etx.isActive()) {
				etx.rollback();
			}
			throw e;
		}
	}

	@Override
	protected void processMessage(Object message) throws Exception {
		Transaction tx = TransactionCoordination.getInstance().getTransaction();
		try {
			MuleMessage muleMessage = createMuleMessage(message, endpoint.getEncoding());
			routeMessage(muleMessage);
		} catch (Exception ex) {
			if (tx != null) {
				tx.setRollbackOnly();
			}
			// rethrow
			throw ex;
		}
	}
}
